package com.atguigu.app

import java.util

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis
import org.json4s.native.Serialization

import collection.JavaConverters._

object SaleDetailApp {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SaleDetailApp")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

    //3.从kafka中获取订单表的数据以及订单明细表的数据
    val orderInfoKafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER, ssc)

    val orderDetailKafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL, ssc)

    //4.将读过来的数转为样例类
    val orderInfoDStream = orderInfoKafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
        //补全create_date&create_hour字段
        orderInfo.create_date = orderInfo.create_time.split(" ")(0)
        orderInfo.create_hour = orderInfo.create_time.split(" ")(1).split(":")(0)

        (orderInfo.id, orderInfo)
      })
    })

    val orderDetailDStream = orderDetailKafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
        (orderDetail.order_id, orderDetail)
      })
    })

    //5.双流join
    //不用join的原因是因为这种方式没有办法获取到join不上的数据
    //    val value: DStream[(String, (OrderInfo, OrderDetail))] = orderInfoDStream.join(orderDetailDStream)

    val joinDStream: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoDStream.fullOuterJoin(orderDetailDStream)


    //6.通过加缓存的方式来解决网络延迟所带来的数据丢失的问题
    val noUserSaleDetailDStream: DStream[SaleDetail] = joinDStream.mapPartitions(partition => {
      //创建redis连接
      val jedis: Jedis = new Jedis("hadoop102", 6379)
      implicit val formats = org.json4s.DefaultFormats
      //创建存放SaleDetail的list集合
      val details: util.ArrayList[SaleDetail] = new util.ArrayList[SaleDetail]()
      partition.foreach { case (orderId, (infoOpt, detailOpt)) =>
        //orderInfo的rediskey
        val infoRedisKey: String = "orderInfo:" + orderId

        //orderDetail的redisKey
        val detailRedisKey: String = "orderDetail:" + orderId

        //a.判断orderInfo数据是否存在
        if (infoOpt.isDefined) {
          //orderInfo数据存在
          //a.2取出orderInfo数据
          val orderInfo: OrderInfo = infoOpt.get
          //a.3判断orderDetail数据是否存在
          if (detailOpt.isDefined) {
            //orderDetail数据存在
            //a.4取出orderDetail数据
            val orderDetail: OrderDetail = detailOpt.get
            //a.5将orderInfo数据和orderDetail数据关联起来
            val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
            details.add(detail)
          }
          //b.1将orderInfo数据写入redis缓存
          //将orderInfo样例类转为Json字符串
          //样例类不能通过toJSONString方法转为JSON字符串，编译错误
          //          JSON.toJSONString(orderInfo)
          val orderInfoJson: String = Serialization.write(orderInfo)
          jedis.set(infoRedisKey, orderInfoJson)

          //给orderInfo数据设置过期时间，防止redis内存不够用
          jedis.expire(infoRedisKey, 100)

          //c.查询orderDetail缓存看是否有能够关联上的数据
          //c.1首先判断redis中是否有对应的key
          if (jedis.exists(detailRedisKey)) {
            //取出orderDetail数据
            val orderDetailStrSet: util.Set[String] = jedis.smembers(detailRedisKey)
            for (elem <- orderDetailStrSet.asScala) {
              //c.2将查询出来的字符串转为样例类
              val orderDetail: OrderDetail = JSON.parseObject(elem, classOf[OrderDetail])
              val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
              details.add(detail)
            }
          }
        } else {
          //orderInfo数据不在
          //d.判断orderDetail数据是否存在
          if (detailOpt.isDefined) {
            //orderDetail数据存在,并取出orderDetail数据
            val orderDetail: OrderDetail = detailOpt.get
            //d.2查询redis中是否有能够关联上的orderInfo的redisKey
            if (jedis.exists(infoRedisKey)) {
              //有能够关联上的orderInfo数据
              //d.3取出能够关联上的OrderInfo数据
              val orderInfoStr: String = jedis.get(infoRedisKey)
              //d.4将获取到的json字符串转为样例类
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoStr, classOf[OrderInfo])
              val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
              //d.5将SaleDetail写入结果集合
              details.add(detail)
            } else {
              //e.redis中没有能够关联上的数据，则把自己（orderDetail）写入缓存
              //e.2将样例类转为JSON字符串
              val orderDetailJsonStr: String = Serialization.write(orderDetail)
              jedis.sadd(detailRedisKey, orderDetailJsonStr)
              //e.3给orderDetail数据设置过期时间
              jedis.expire(detailRedisKey, 100)
            }
          }
        }
      }
      jedis.close()
      details.asScala.toIterator
    })
    //7.反查缓存关联用户信息
    val SaleDetailDStream: DStream[SaleDetail] = noUserSaleDetailDStream.mapPartitions(partition => {
      //创建redis连接
      val jedis: Jedis = new Jedis("hadoop102", 6379)
      val saleDetails: Iterator[SaleDetail] = partition.map(saleDetail => {
        //用户表的Rediskey
        val userInfoRedisKey: String = "userInfo:" + saleDetail.user_id
        val userInfoStr: String = jedis.get(userInfoRedisKey)
        //将查询出来的数据转为样例类
        val userInfo: UserInfo = JSON.parseObject(userInfoStr, classOf[UserInfo])

       /* if (jedis.exists(userInfoRedisKey)) {
        //查询数据并关联
        }else{
          Thread.sleep(20)
          //查询数据并关联
        }*/

        //关联数据
        saleDetail.mergeUserInfo(userInfo)
        saleDetail
      })
      jedis.close()
      saleDetails
    })
    SaleDetailDStream.print()

    //8.将数据写入ES
    SaleDetailDStream.foreachRDD(rdd=>{
      rdd.foreachPartition(partition=>{
        val list: List[(String, SaleDetail)] = partition.toList.map(saleDetail => {
          (saleDetail.order_detail_id, saleDetail)
        })
       MyEsUtil.insertBulk(GmallConstants.ES_INDEX_SALEDETAIL+"210927",list)
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }

}
